package local.sqlsubmit;

import local.sqlsubmit.cli.CliOptions;
import local.sqlsubmit.cli.CliOptionsParser;
import local.sqlsubmit.cli.SqlCommandParser;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.core.execution.JobClient;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.SqlParserException;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;


import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

/**
 * @date : 2020-7-22 10:12:01
 * @desc : 提交SQL的命令
 */
public class SqlSubmit {

    public static void main(String[] args) throws Exception {
        final CliOptions options = CliOptionsParser.parseClient(args);
        SqlSubmit submit = new SqlSubmit(options);
        submit.run();
    }

    // --------------------------------------------------------------------------------------------

    private String sqlFilePath;
    private TableEnvironment tEnv;

    private SqlSubmit(CliOptions options) {
        this.sqlFilePath = options.getSqlFilePath();
    }


    private void run() throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                900, // 尝试重启次数
                Time.of(20, TimeUnit.SECONDS) // 延迟时间间隔
        ));

        this.tEnv = StreamTableEnvironment.create(
                environment,
                EnvironmentSettings.
                        newInstance().
                        inStreamingMode().
                        useBlinkPlanner().
                        build()
        );

        String [] sqlFilePaths = sqlFilePath.split(",");
        List<String> sql = new ArrayList<String>();
        for (int i = 0; i < sqlFilePaths.length; i++) {
            List<String> tmp = Files.readAllLines(Paths.get(sqlFilePaths[i]));
            if(tmp.size()>0){
                sql.addAll(tmp);
            }
        }
        List<SqlCommandParser.SqlCommandCall> calls = SqlCommandParser.parse(sql);
        if (calls.size() == 0) {
            //no sql to execute
            throw new RuntimeException("There is no sql statement to execute,please check your sql file: " + sqlFilePath);
        }
        for (SqlCommandParser.SqlCommandCall call : calls) {
//            System.out.println(call.command.toString());
            callCommand(call);
        }
    }

    // --------------------------------------------------------------------------------------------

    private void callCommand(SqlCommandParser.SqlCommandCall cmdCall) {
        switch (cmdCall.command) {
            case SET:
                callSet(cmdCall);
                break;
            case CREATE_TABLE:
                callCreateTable(cmdCall);
                break;
            case CREATE_FUNCTION:
                callCreateFunction(cmdCall);
                break;
            case CREATE_TEMP_VIEW:
                callCreateTempView(cmdCall);
                break;
            case INSERT_INTO:
                callInsertInto(cmdCall);
                break;
            default:
                throw new RuntimeException("Unsupported command: " + cmdCall.command);
        }
    }

    private void callSet(SqlCommandParser.SqlCommandCall cmdCall) {
        String key = cmdCall.operands[0];
        String value = cmdCall.operands[1];
        tEnv.getConfig().getConfiguration().setString(key, value);
        System.out.println("set " + key + "->" + value + " ok");
    }

    private void callCreateTable(SqlCommandParser.SqlCommandCall cmdCall) {
        String ddl = cmdCall.operands[0];
        try {
            tEnv.executeSql(ddl);
        } catch (SqlParserException e) {
            throw new RuntimeException("SQL parse failed:\n" + ddl + "\n", e);
        }
        String tableName = ddl.split("\\s+")[2];
        System.out.println("crate table " + tableName + " ok");
    }

    private void callCreateTempView(SqlCommandParser.SqlCommandCall cmdCall) {
        String ddl = cmdCall.operands[0];
        try {
            tEnv.executeSql(ddl);
        } catch (SqlParserException e) {
            throw new RuntimeException("SQL parse failed:\n" + ddl + "\n", e);
        }
        String tableName = ddl.split("\\s+")[2];
        System.out.println("crate temporrary " + tableName + " ok");
    }


    private void callCreateFunction(SqlCommandParser.SqlCommandCall cmdCall) {
        String ddl = cmdCall.operands[0];
        try {
            tEnv.executeSql(ddl);
        } catch (SqlParserException e) {
            throw new RuntimeException("SQL parse failed:\n" + ddl + "\n", e);
        }
        String functionName = ddl.split("\\s+")[2];
        System.out.println("crate function " + functionName + " ok");
    }

    private void callInsertInto(SqlCommandParser.SqlCommandCall cmdCall) {
        String dml = cmdCall.operands[0];
        Optional<JobClient> jobClient;
        try {
            TableResult result = tEnv.executeSql(dml);
            jobClient = result.getJobClient();
        } catch (SqlParserException e) {
            throw new RuntimeException("SQL parse failed:\n" + dml + "\n", e);
        }

        if (jobClient.isPresent()) {
            JobID jobID = jobClient.get().getJobID();
            System.out.println("job submit ok,JobId: " + jobID);
        }

    }
}
